package com.atguigu.edu.realtime.app.dwd.db;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Map;

public class DwdTradeOrderSource {
    public static void main(String[] args) throws Exception {
        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);



        // TODO 3. 从 Kafka 读取主流数据
        String topic = "dwd_traffic_page_log";

        String groupId = "dwd_trade_order_source_group";
        DataStreamSource<String> kafkaStrDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, groupId));


        //TODO 4. 转换数据结构
        SingleOutputStreamOperator<JSONObject> jsonObj = kafkaStrDS.map(JSONObject::parseObject);


       // jsonObj1.print("1");
        //jsonObj2.print("2");
        //TODO 5. 过滤数据
        SingleOutputStreamOperator<JSONObject> filterDS = jsonObj.filter(
                new FilterFunction<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        return "order".equals(value.getJSONObject("page").getString("page_id"));
                    }
                }
        );

        SingleOutputStreamOperator<String> mapDs = filterDS.map(JSONAware::toJSONString);


        mapDs.addSink(MyKafkaUtil.getKafkaProducer("dwd_trade_order_source"));


        env.execute();
    }

}
